package com.its.common.kafka.producer;

import com.its.common.kafka.config.KafkaProducerProperties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;

import java.util.Properties;

/**
 * @Author xiaxp
 * @Date 2021-11-09 21:58
 * @Description
 *
 *  使用生产者的方式

       @Autowired
       private Producer producer;

       ProducerRecord<String,Object> record = new ProducerRecord<>(topicName,key,reportData);


       1、Kafka Producer是线程安全的，建议多线程复用，如果每个线程都创建，出现大量的上下文切换或争抢的情况，影响Kafka效率
       2、Kafka Producer的key是一个很重要的内容：
         2.1 我们可以根据Key完成Partition的负载均衡
         2.2 合理的Key设计，可以让Flink、Spark Streaming之类的实时分析工具做更快速处理

       3、ack - all， kafka层面上就已经有了只有一次的消息投递保障，但是如果想真的不丢数据，最好自行处理异常

        try{
            producer.send(record);
        }catch(Exception e){
            // 将数据加入重发队列， redis，es，...
        }
 *
 *
 */
@ConditionalOnProperty(prefix = "its.kafka.producer", name = "enabled", havingValue = "true")
@EnableConfigurationProperties({KafkaProducerProperties.class})
public class KafkaProducerProvider {

    @Autowired
    private KafkaProducerProperties kafkaProducerProperties;

    @Bean
    public Producer kafkaProducer(){
        Properties properties = kafkaProducerProperties.buildProperties();
        // Producer的主对象
        Producer<String,String> producer = new KafkaProducer<>(properties);
        return producer;
    }
}
